Summary

AsyncExecutor can cause a run to be stuck in STARTED indefinitely when the last in-flight
step is parked for a delayed retry.

_async_execution_iterator drives scheduling off the event stream and blocks
unconditionally on recv_stream.receive(). When the final in-flight step emits
STEP_UP_FOR_RETRY under a RetryPolicy(delay > 0), Dagster parks it in
ActiveExecution._waiting_to_retry until the backoff elapses. During that window:

• get_steps_to_execute() returns [] (the step is promoted to executable inside
  _update() only once time.time() >= at_time), so no worker is started;
• the worker that emitted the retry event has already finished, so nothing is alive
  to send another event;
• recv_stream.receive() blocks forever, while active.is_complete stays False
  because it counts _waiting_to_retry.

The stock in_process / multiprocess executors avoid this by polling with
active.sleep_til_ready() / sleep_interval() instead of blocking on an event.

Fix

Bound the receive with anyio.move_on_after(active.sleep_interval()) when something
time-based is pending, and re-poll on timeout so get_steps_to_execute() →
_update() promotes the now-ready retry step. When sleep_interval() == 0 (nothing
parked), move_on_after(None) is an unbounded wait — identical to the previous
behavior on the hot path.

This mirrors the stock executors' sleep_til_ready() polling. The timeout also bounds
receive() on the pending concurrency-claim path, which is likewise strictly closer to
the stock executors' behavior. The removed try/except anyio.EndOfStream: raise was a
no-op (EndOfStream cannot fire while the orchestrator's send_stream is open); it is
still handled by _drain_events after the loop.

Test

test_retry_policy.py: a single-op job that fails on its first attempt and succeeds on
the retry under RetryPolicy(max_retries=1, delay=0.5). A single op is the minimal
reproducer because it is by definition the last in-flight step. Guarded with
@pytest.mark.timeout(30) (via pytest-timeout, added as a dev dependency) so the
pre-fix hang surfaces as a fast failure instead of wedging the suite. The test asserts
positive evidence that the retry actually ran (the op's returned attempt count and a
per-test attempts file), not merely result.success.

Changes

• executor.py — bound recv_stream.receive() with anyio.move_on_after(sleep_interval()), re-poll on timeout.
• test_retry_policy.py — new timeout-guarded regression test.
• pyproject.toml / uv.lock — add pytest-timeout dev dependency (and de-dupe a duplicate pytest entry).